package mapreduce;

import java.nio.ByteBuffer;
import java.util.Vector;

public class DynamicNumberFormat extends BytesFormat {

	public DynamicNumberFormat() {
	}
	
	public DynamicNumberFormat(BytesFormat baseFormat){
		setBaseFormat(baseFormat);
	}
	
	public void setBaseFormat(BytesFormat baseFormat){
		m_baseFormat = baseFormat;
	}

	protected byte[] format(Object object) {
		Vector<byte[]> outBytesVec = new Vector<byte[]>();
		Object[] objs = (Object[]) object;
		int typeNum = objs.length;
		int byteNum = 0;
		for(int i = 0 ; i < typeNum ; i ++){
			byte[] outBytes = m_baseFormat.format(objs[i]);
			outBytesVec.add(outBytes);
			byteNum += outBytes.length;
		}
		ByteBuffer outBuffer = ByteBuffer.allocate(byteNum + typeNum);
		for (int i = 0; i < typeNum; i++) {
			byte[] bytes = outBytesVec.elementAt(i);
			outBuffer.put(bytes);
			outBuffer.put((byte) ';');
		}
		return outBuffer.array();
	}

	private BytesFormat m_baseFormat;

	@Override
	public Object decode(ByteBuffer buffer) {
		m_baseFormat = BytesFormat.getFormt(buffer.getInt());
		Vector<Object> objectVec = new Vector<Object>();
		while (true) {
			Object object = m_baseFormat.decode(buffer);
			objectVec.add(object);
			if(buffer.position() >= buffer.limit()){
				break;
			}
		}
		return objectVec.toArray();
	}

	@Override
	public ByteBuffer encode(Object object, ByteBuffer buffer) {
		buffer = BufferChecker.check(buffer, 4);
		buffer.putInt(m_baseFormat.getCode());
		Object[] objects = (Object[]) object;
		int objectNum = objects.length;
		for(int i = 0 ; i < objectNum ; i ++){
			buffer = m_baseFormat.encode(objects[i], buffer);
		}
		return buffer;
	}

	@Override
	public BytesFormat getFormat() {
		return new DynamicNumberFormat();
	}

	@Override
	protected int getCode() {
		return 10;
	}

}
